#include "xlcomm/base/async_log.h"

#include <functional>

namespace xlcomm {

const int kReserveSize = 8;

AsyncLog::AsyncLog(const char* basename, int64_t roll_size)
    : sync_log_(basename, roll_size, false),
      flush_interval_second_(1),
      start_latch_(1),
      cur_buffer_(new Buffer),
      next_buffer_(new Buffer),
      running_(false) {
  buffers_queue_.reserve(kReserveSize);
  Start();
}

AsyncLog::~AsyncLog() { Stop(); }

void AsyncLog::Append(const char* msg, size_t len) {
  std::lock_guard<std::mutex> guard(mutex_);
  if (cur_buffer_->AvailSize() > len) {  // 当前可用，直接用
    cur_buffer_->Append(msg, len);
  } else {
    buffers_queue_.push_back(std::move(cur_buffer_));
    if (next_buffer_ != nullptr) {
      cur_buffer_ = std::move(next_buffer_);  // 备用可用，把备用的拿来用
    } else {
      cur_buffer_.reset(new Buffer);  // 备用也不可用，重新分配空间（极少发生）
    }
    cur_buffer_->Append(msg, len);
    condition_.notify_one();
  }
}

void AsyncLog::Flush() {
  std::lock_guard<std::mutex> guard(mutex_);
  sync_log_.Flush();
}

void AsyncLog::Start() {
  XLCOMM_ASSERT0(!running_);
  running_ = true;
  write_thread_ = std::thread(std::bind(&AsyncLog::ThreadProc, this));
  start_latch_.WaitZero();
}

void AsyncLog::Stop() {
  XLCOMM_ASSERT0(running_);
  running_ = false;
  condition_.notify_one();
  write_thread_.join();
}

void AsyncLog::ThreadProc() {
  BufferPtr new_buffer1(new Buffer);
  BufferPtr new_buffer2(new Buffer);
  XLCOMM_ASSERT0(running_);
  start_latch_.CountDown();
  // 如果AsyncLog对象生命周期太短，有可能进入不了while循环
  while (running_) {
    std::vector<BufferPtr> buffers_to_write;
    buffers_to_write.reserve(kReserveSize);
    {
      std::unique_lock<std::mutex> lock(mutex_);
      if (buffers_queue_.empty()) {
        // 等待缓冲队列有数据可写、停止、超时
        condition_.wait_for(lock, std::chrono::seconds(flush_interval_second_));
      }
      if (cur_buffer_->used() > 0) {
        // 同时把cur_buffer_放入缓冲队列，并替换cur_buffer_为新的空间
        buffers_queue_.push_back(std::move(cur_buffer_));
        cur_buffer_ = std::move(new_buffer1);
      }
      if (next_buffer_ == nullptr) {
        next_buffer_ = std::move(new_buffer2);
      }
      buffers_to_write.swap(buffers_queue_);
    }

    // 缓冲队列短时间内累积太多，可能是死循环了，丢弃后产生的无用日志
    if (buffers_to_write.size() > kReserveSize * 2) {
      char str[256];
      snprintf(str, sizeof(str),
               "Dropped log messages at %s, dropped %zd larger buffers!\n",
               DateTime::Format(DateTime::Now()).c_str(),
               buffers_to_write.size() - 2);
      fputs(str, stderr);
      sync_log_.Append(str, strlen(str));
      buffers_to_write.erase(buffers_to_write.begin() + 2,
                             buffers_to_write.end());
    }

    // 将日志信息写入文件
    for (auto& buff : buffers_to_write) {
      sync_log_.Append(buff->data(), buff->used());
    }

    if (buffers_to_write.size() > 2) {
      // 释放后分配的堆内存，避免内存碎片
      buffers_to_write.resize(2);
    }

    // 循环利用之前分配的堆内存
    if (new_buffer1 == nullptr) {
      XLCOMM_ASSERT0(!buffers_to_write.empty());
      new_buffer1 = std::move(buffers_to_write.back());
      buffers_to_write.pop_back();
      new_buffer1->Reset();
    }
    if (new_buffer2 == nullptr) {
      XLCOMM_ASSERT0(!buffers_to_write.empty());
      new_buffer2 = std::move(buffers_to_write.back());
      buffers_to_write.pop_back();
      new_buffer2->Reset();
    }
    buffers_to_write.clear();

    sync_log_.Flush();
  }
  sync_log_.Flush();
}

}  // namespace xlcomm